package com.yl.flink.processor;

import com.yl.entity.MultiDataEntity;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.List;

/**
 * @author wlf
 * @since 2022/8/26
 */
public class MaxReduceFunc implements ReduceFunction<MultiDataEntity> {

    // 计算所用的指标序号
    private final int qc;

    public MaxReduceFunc(List<String> calRefers) {
        this.qc = Integer.parseInt(calRefers.get(0));
    }

    /**
     * 获取最大值
     *
     * @param e1 前一个流元素
     * @param e2 后一个流元素
     */
    @Override
    public MultiDataEntity reduce(MultiDataEntity e1, MultiDataEntity e2) throws Exception {
        // 返回数据较大的流元素
        MultiDataEntity multiDataEntity = e1.getRawValues().get(qc) > e2.getRawValues().get(qc) ? e1 : e2;
        // 以窗口起始时间作为数据时间
        multiDataEntity.setPickTime(e1.getPickTime());
        // 设置计算后的最大值
        multiDataEntity.setCalValues(Tuple2.of(e1.getCalValues().f0, multiDataEntity.getRawValues().get(qc)));
        return multiDataEntity;
    }
}
